package com.atguigu.flink.datastramapi.transform;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/2/25
 *
 *      吃东西前，需要洗手，吃完后，需要擦嘴。
 *
 *          有些处理逻辑只需要在任务开始之前执行一次和结束后执行一次，需要使用
 *              RichFunction的子类。
 *
 *                   RichFunction 中提供了一组生命周期方法。
 *                          open():   Task初始化时，做事
 *                          close():  Task终止时，做事
 *
 *             编写一段逻辑，既要有MapFunction的功能，还要有RichFunction的功能。
 *
 *             基本上使用的算子都提供了 RichXXXFunction。
 *                  常见的场景： 获取数据库连接。
 *
 */
public class Demo4_RichFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataStreamSource<String> ds = env.fromElements("苹果", "饼干", "猪脚饭");

        ds.map(new MyRichMapFunction()).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    private static class MyRichMapFunction extends RichMapFunction<String,String>{

        //Task出生时执行一次
        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("先洗手");
        }

        //Task终止时执行一次
        @Override
        public void close() throws Exception {
            System.out.println("擦嘴");
        }

        @Override
        public String map(String x) throws Exception {

            System.out.println("吃:"+x);

            return "吃剩下的:"+x;

        }
    }
}
